package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Code04MapPartitions {
  def main(args: Array[String]): Unit = {

    /**
     * Spark算子:
     *    mapPartitions作用：和Map作用相似，
     *        但是mapPartitions是根据分区确定其执行的次数，每个分区执行一次当前函数
     *        会将当前分区中的所有数据，包装成一个 Iterator 再由迭代器进行做对应分区中的数据处理
     *        并且返回值要求为一个 Iterator
     *
     *   应用场景:
     *
     *
     */



    val sc = new SparkContext(new SparkConf().setMaster("local").setAppName("Code04MapPartitions"))

    val stuDataRDD: RDD[String] = sc.textFile("spark_code/data/students.txt",5)

    stuDataRDD
      .map(line => {
        println("map执行了一次...")
        val splitRes: Array[String] = line.split(",")
        (splitRes(0), splitRes(1), splitRes(2), splitRes(3), splitRes(4))
      })
      //
      .mapPartitions(
        iterator => {
          println("mapPartitions 执行了一次...")
          val mapResIter: Iterator[(String, String)] = iterator
            // 对当前分区迭代器数据处理
            .map {
              case (id, name, age, gender, clazz) => {
                (id, clazz)
              }
            }
          // 返回一个 mapResIter
          mapResIter
        }
      )
      .groupBy(_._2)
      .map{
        case (clazz,oneClazzStu) => (clazz,oneClazzStu.size)
      }
      .foreach(println)


  }
}
